package com.example.kafka.demo;


import com.example.kafka.service.Consumer;
import com.example.kafka.service.Producer;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;

/**
 * @application:
 * @author: LiYuhang
 * @Date: 2021/6/26 14:45
 * @version: 0.1
 * @Description :
 * @Modified By:
 */
@Slf4j
public class ConsumerDemo1 implements Consumer {

    private final KafkaConsumer<String, String> consumer;
    private final String topic;


    public ConsumerDemo1(String topic,String groupID) {
        this.consumer = new KafkaConsumer<String, String>(Consumer.getDefaultConfig(groupID));
        this.topic = topic;
    }

    public void run() {
        this.consumer.subscribe(Collections.singleton(this.topic));
        while (true) {
            ConsumerRecords<String, String> records =
                    consumer.poll(Duration.ofMillis(100));
            if (records.isEmpty()) {
                log.warn("未读取到消息，sleep 500");
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {

                }
            } else {
                for (ConsumerRecord<String, String> record : records) {
                    log.info("run:{},{}",record.value(),record);
                }
            }
        }
    }


    /**
     * 流式增强
     * @param producer
     */
    public void run(Producer producer) {
        this.consumer.subscribe(Collections.singleton(this.topic));
        while (true) {
            ConsumerRecords<String, String> records =
                    consumer.poll(Duration.ofMillis(100));
            if (records.isEmpty()) {
                log.warn("未读取到消息，sleep 500");
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {

                }
            } else {
                for (ConsumerRecord<String, String> record : records) {
                    String value = record.value();
                    log.info("run:{},{}",value,record);
                    Integer integer = Integer.valueOf(value.substring(value.length() - 2,value.length()));
                    if( integer % 2 == 0) {
                        log.warn("valid data:{}",integer);
                        producer.process(String.valueOf(integer));
                    }
                }
            }
        }
    }

    public static void main(String[] args) {
        //test
        //"topic.group3"
        ConsumerDemo1 demo1 = new ConsumerDemo1("test","topic.group2");
        demo1.run();
    }


}
